package com.wanglei.project

import org.apache.spark.{SparkConf, SparkContext}

object Spark01_top10 {

  def main(args: Array[String]): Unit = {
    /**
     * top10热门品类
     * 品类id -> (点击数量，下单数量，支付数量)
     */

    val conf = new SparkConf().setMaster("local[2]").setAppName("top")
    val sc = new SparkContext(conf)

    // 
    val actionRdd = sc.textFile("datas/user_visit_action.txt")
    // 品类id->点击数量
    val clickCntRdd = actionRdd
      .filter(
        action => {
          val datas = action.split("_")
          datas(6) != "-1"
        }
      )
      .map(
        action => {
          val datas = action.split("_")
          (datas(6), 1)
        }
      )
      .reduceByKey(_ + _)


    // 品类id -> 下单数量
    val orderCntRdd = actionRdd
      .filter(
        action => {
          val datas = action.split("_")
          datas(8) != "null"
        }
      )
      .flatMap(
        action => {
          val datas = action.split("_")
          val cids = datas(8).split(",")
          cids.map(id => (id, 1))
        }
      )
      .reduceByKey(_ + _)

    // 品类id -> 支付数量
    val payCntRdd = actionRdd
      .filter(
        action => {
          val datas = action.split("_")
          datas(11) != "null"
        }
      )
      .flatMap(
        action => {
          val datas = action.split("_")
          val cids = datas(11).split(",")
          cids.map(id => (id, 1))
        }
      )
      .reduceByKey(_ + _)

    // 将三个rdd连接在一起
    // (品类id,(点击数量，下单数量，支付数量))
    // cogroup
    val coGroupedRDD = clickCntRdd.cogroup(orderCntRdd, payCntRdd)
    coGroupedRDD.mapValues {
      case (click, order, pay) => {
        var clickCnt = 0
        val iterator1 = click.iterator
        if (iterator1.hasNext) {
          clickCnt = iterator1.next()
        }

        var orderCnt = 0
        val iterator2 = order.iterator
        if (iterator2.hasNext) {
          orderCnt = iterator2.next()
        }


        var payCnt = 0
        val iterator3 = pay.iterator
        if (iterator3.hasNext) {
          payCnt = iterator3.next()
        }

        (clickCnt, orderCnt, payCnt)

      }
    }
      .sortBy(_._2, ascending = false).take(10)
      .foreach(println(_))

    //
    sc.stop()
  }

}
